package events

import (
	"context"
	"github.com/IBM/sarama"
	"jk-time/webook/follow/cache"
	"jk-time/webook/pkg/canalx"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/saramax"
	"time"
)

type MySQLBinlogComsumer struct {
	client sarama.Client
	l      logger.Logger
	cache  cache.FollowCache
}

func (k *MySQLBinlogComsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("follow_binlog", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{"example"},
			saramax.NewHandler[canalx.Message[Follow]]("follow_binlog", k.Consume))
		if err != nil {
			k.l.Error("退出了消费循环,异常", logger.Error(err))
		}
	}()
	return nil
}

func (k *MySQLBinlogComsumer) Consume(msg *sarama.ConsumerMessage, cmsg canalx.Message[Follow]) error {
	if cmsg.Table != "follow" || (cmsg.Type != "INSERT" && cmsg.Type != "UPDATE") {
		return nil
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	for _, data := range cmsg.Data {
		var err error
		switch data.Status {
		case FollowRelationStatusActive:
			err = k.cache.UpdateStaticsInfo(ctx, data.Follower, data.Followee, 1)
		case FollowRelationStatusInactive:
			err = k.cache.UpdateStaticsInfo(ctx, data.Follower, data.Followee, -1)
		default:
			//记录日志
		}
		if err != nil {
			return err
		}
	}
	return nil
}

type Follow struct {
	Id       int64 `gorm:"autoIncrement,primaryKey"`
	Follower int64 `gorm:"uniqueIndex:er_ee"` //去重
	Followee int64 `gorm:"uniqueIndex:er_ee;index"`
	//软删除
	Status uint8

	Ctime int64
	Utime int64
}

const (
	FollowRelationStatusUnknown uint8 = iota
	FollowRelationStatusActive
	FollowRelationStatusInactive
)
